package com.ermao.mymes.handler.receiver;

import com.alibaba.fastjson.JSON;
import com.ermao.mymes.common.domain.AnchorInfo;
import com.ermao.mymes.common.domain.LogParam;
import com.ermao.mymes.common.domain.TaskInfo;
import com.ermao.mymes.common.enums.AnchorState;
import com.ermao.mymes.handler.pending.Task;
import com.ermao.mymes.handler.utils.GroupIdMappingUtils;
import com.ermao.mymes.handler.pending.TaskPendingHolder;
import com.ermao.mymes.support.utils.LogUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Scope;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Optional;

/**
 * @author Ermao
 * Date: 2022/5/8 11:10
 */
@Component
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class Receiver {
	private static final String LOG_BIZ_TYPE = "Receiver#consumer";
	@Autowired
	private ApplicationContext context;

	@Autowired
	private TaskPendingHolder taskPendingHolder;

	@KafkaListener(topics = "${mymes.topic.name}")
	public void consumer(ConsumerRecord<?, String> consumerRecord, @Header(KafkaHeaders.GROUP_ID) String topicGroupId) {
		Optional<String> kafkaMessage = Optional.ofNullable(consumerRecord.value());
		if (kafkaMessage.isPresent()) {

			List<TaskInfo> taskInfoLists = JSON.parseArray(kafkaMessage.get(), TaskInfo.class);
			String messageGroupId = GroupIdMappingUtils.getGroupIdByTaskInfo(taskInfoLists.get(0));

			/**
			 * 每个消费者组 只消费 他们自身关心的消息
			 * TODO 理解
			 */
			if (topicGroupId.equals(messageGroupId)) {
				for (TaskInfo taskInfo : taskInfoLists) {
					LogUtils.print(LogParam.builder().bizType(LOG_BIZ_TYPE).object(taskInfo).build(), AnchorInfo.builder().ids(taskInfo.getReceiver()).businessId(taskInfo.getBusinessId()).state(AnchorState.RECEIVE.getCode()).build());
					Task task = context.getBean(Task.class).setTaskInfo(taskInfo);
					taskPendingHolder.route(topicGroupId).execute(task);
				}
			}
		}

	}
}
